package cloud.xiguapi.ubas.analysis.uv.model;

import cloud.xiguapi.ubas.model.PageViewCount;
import cloud.xiguapi.ubas.model.UserBehavior;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

/**
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-19 上午 09:55
 * desc:
 */
public class BloomFilterUvCountResultProcess extends ProcessAllWindowFunction<UserBehavior, PageViewCount, TimeWindow> {

    /**
     * Jedis的连接地址
     */
    public static final String JEDIS_HOST = "localhost";

    /**
     * Jedis的连接端口
     */
    public static final int JEDIS_PORT = 6379;

    /**
     * 布隆过滤器的容量
     */
    public static final int BLOOM_FILTER_CAP = 1 << 29;

    /**
     * redis中hash表的名称
     */
    public static final String COUNT_HASH_NAME = "uv_count";

    /**
     * Jedis客户端
     */
    private transient Jedis client;

    /**
     * 自定义UV Count布隆过滤器
     */
    private transient UvCountResultBloomFilter filter;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化Jedis客户端和布隆过滤器
        client = new Jedis(JEDIS_HOST, JEDIS_PORT);
        // 处理1亿个数据, 用64MB大小的位图, 需要的容量为2的29次方.
        // 使用位移操作更快捷
        filter = new UvCountResultBloomFilter(BLOOM_FILTER_CAP);
    }

    @Override
    public void close() throws Exception {
        client.close();
    }

    @Override
    public void process(Context context, Iterable<UserBehavior> elements, Collector<PageViewCount> out) throws Exception {
        // 将位图和窗口count值全部存入redis, 用windowEnd为Key
        Long windowEnd = context.window().getEnd();
        String bitmapKey = windowEnd.toString();

        // 把count值存成一张hash表
        String countKey = windowEnd.toString();

        // 取出当前userId
        Long userId = elements.iterator().next().getUserId();

        // 计算位图中的offset, seed选用一个质数
        Long offset = filter.hashCode(userId.toString(), 61);

        // 用redis的getbit命令判断对应位置的值
        Boolean isExist = client.getbit(bitmapKey, offset);

        if (!isExist) {
            // 如果不存在, 对应位图位置置1
            client.setbit(bitmapKey, offset, true);
        }

        // 更新redis中保存的count值
        // 定义初始值uvCount
        long uvCount = 0L;

        String uvCountString = client.hget(COUNT_HASH_NAME, countKey);

        if (uvCountString != null && !"".equals(uvCountString)) {
            uvCount = Long.parseLong(uvCountString);
        }

        // 将累加后的值放入redis
        client.hset(COUNT_HASH_NAME, countKey, String.valueOf(uvCount + 1));

        out.collect(PageViewCount.builder()
                .url("uv")
                .windowEnd(windowEnd)
                .count(uvCount + 1)
                .build());
    }
}